package com.hibase.core.interfacetask.service.impl;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.github.pagehelper.PageInfo;
import com.hibase.common.exception.HibaseAssert;
import com.hibase.common.exception.HibaseException;
import com.hibase.common.util.EntityUtil;
import com.hibase.core.interfacetask.constant.TaskDealStatus;
import com.hibase.core.interfacetask.entity.MdsInterfaceTask;
import com.hibase.core.interfacetask.entity.MdsInterfaceType;
import com.hibase.core.interfacetask.entity.ResponseInfo;
import com.hibase.core.interfacetask.entity.vo.MdsInterfaceTaskSearchVO;
import com.hibase.core.interfacetask.entity.vo.MdsInterfaceTaskThirdCallVO;
import com.hibase.core.interfacetask.mapper.MdsInterfaceTaskMapper;
import com.hibase.core.interfacetask.service.IMdsInterfaceTaskService;
import com.hibase.core.interfacetask.service.MdsInterfaceTypeService;
import com.hibase.core.interfacetask.util.HibaseTaskUtil;
import com.hibase.core.mq.component.HibaseQueueUtil;
import com.hibase.core.mybatis.service.BaseServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.web.client.RestTemplate;

import java.util.List;
import java.util.Map;


/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author hufeng
 * @since 2019-04-29
 */
@Service
@Transactional(rollbackFor = Exception.class)
@Slf4j
public class MdsInterfaceTaskServiceImpl extends BaseServiceImpl<MdsInterfaceTaskMapper, MdsInterfaceTask> implements IMdsInterfaceTaskService {


    @Autowired
    private MdsInterfaceTypeService mdsInterfaceTypeService;


    //todo
    @Autowired(required = false)
    RestTemplate restTemplate;

    @Autowired
    private HibaseQueueUtil hibaseQueueUtil;

    @Autowired
    private HibaseTaskUtil hibaseTaskUtil;

    /**
     * 查询未处理的任务
     */
    @Override
    public List<MdsInterfaceTask> selectNotDealList(Map<String, Object> map) {

        return super.baseMapper.selectNotDealList(map);
    }

    /**
     * 保存任务处理状态
     */
    private void saveTaskStatus(String status, String id, String responseData) {

        MdsInterfaceTask mdsInterfaceTask = super.getById(id);

        mdsInterfaceTask.setDealStatus(status);
        mdsInterfaceTask.setExecTime(DateUtil.date());
        mdsInterfaceTask.setResponseData(responseData);

        super.store(mdsInterfaceTask);
    }

    /**
     * 改变接口任务表记录
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveMdsInterfaceTask(MdsInterfaceTask entity) {
        if (StrUtil.isNotBlank(entity.getId())){
            super.store(entity);
        }else {
            // 新增接口任务
            String businessTypeCode = entity.getBusinessTypeCode();
            MdsInterfaceType interfaceType = mdsInterfaceTypeService.getOne(new QueryWrapper<MdsInterfaceType>().eq("business_type_code", businessTypeCode));
            HibaseAssert.notNull(interfaceType,"未找到对于的Type");
            entity.setReceiverSysUrl(interfaceType.getUrl());
            entity.setBusinessType(interfaceType.getBusinessTypeName());
            entity.setSenderSys(interfaceType.getSenderSys());
            entity.setReceiverSys(interfaceType.getReceiverSys());
            entity.setBusinessId(entity.getBusinessId());
            entity.setBusinessCode(entity.getBusinessCode());
            entity.setDealStatus(TaskDealStatus.TASK_DEAL_OPEN.getName());
            super.store(entity);
        }
    }



    /**
     * 查询接口任务表详情
     */
    @Override
    public MdsInterfaceTask viewMdsInterfaceTask(String id) {

        MdsInterfaceTask entity = super.getById(id, true);

        return entity;
    }

    /**
     * 立刻执行接口
     * @param ids
     * @return
     */
    @Override
    public String executeTask(List<String> ids) {
        List<MdsInterfaceTask> list = (List<MdsInterfaceTask>) listByIds(ids);
        HibaseAssert.isTrue(list.stream().allMatch(p->StrUtil.equalsAny(p.getDealStatus(),TaskDealStatus.TASK_DEAL_FAIL.getName(),
                TaskDealStatus.TASK_DEAL_READY.getName(),TaskDealStatus.TASK_DEAL_OPEN.getName())),"请选择未成功状态得数据");
        StringBuilder result = new StringBuilder();
        for (MdsInterfaceTask task: list) {
            if (task.getRetryCount() == null){
                task.setRetryCount(0);
            }
            result.append(asyncCall(task));
        }
        return result.toString();
    }

    /**
     * 异步调用接口
     * 1，信息入库
     * 2，写入消息队列的指定topic中
     * @param interfaceTaskVO
     * @param topicName
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public String asyncCall(MdsInterfaceTaskThirdCallVO interfaceTaskVO, String topicName) {

        MdsInterfaceTask entity = new MdsInterfaceTask();

        //1，缓存中获取type表信息
        MdsInterfaceType interfaceType = this.getInterfaceType(interfaceTaskVO);
        if (StrUtil.isBlank(interfaceType.getUrl())) {
            log.info("URL为空: {}",interfaceType);
            throw new HibaseException(999,"URL为空");
        }
        //2，拼装task表记录并入库
        this.saveTask(interfaceTaskVO, entity, interfaceType);
        //3，发送消息
        //不传管道名称则默认为正常管道，否则为补偿管道
//        if (StrUtil.isBlank(topicName)) {
//            topicName = interfaceType.getTopicNameNormal();
//        }
        TransactionSynchronizationManager.registerSynchronization( //在一个有事务的方法中，等事务提交后调另外一个方法可以用
                new TransactionSynchronizationAdapter() {
                    @Override
                    public void afterCommit() {
                        kafkaSendMsg(entity, interfaceType.getTopicNameNormal());
                    }
                }
        );
        return "null";
    }
    /**
     * 异步调用接口
     * 1，信息入库
     * 2，写入消息队列的指定topic中
     * @param task
     */
    @Override
    public String asyncCall(MdsInterfaceTask task) {


         MdsInterfaceTaskThirdCallVO callVO = EntityUtil.transform(task, MdsInterfaceTaskThirdCallVO.class);
         //1，缓存中获取type表信息
        MdsInterfaceType interfaceType = this.getInterfaceType(callVO);
        if (StrUtil.isBlank(interfaceType.getUrl())) {
            log.info("URL为空: {}",interfaceType);
            throw new HibaseException(999,"URL为空");
        }
         //2，发送消息
         //不传管道名称则默认为正常管道，否则为补偿管道
         String topicName = interfaceType.getTopicNameException();
         task.setReceiverSysUrl(interfaceType.getUrl());

        TransactionSynchronizationManager.registerSynchronization( //在一个有事务的方法中，等事务提交后调另外一个方法可以用
                new TransactionSynchronizationAdapter() {
                    @Override
                    public void afterCommit() {
                        kafkaSendMsg(task, topicName);
                    }
                }
        );
         //设置deal_status状态为准备状态
        task.setDealStatus(TaskDealStatus.TASK_DEAL_READY.getName());
        super.saveOrUpdate(task);
        return task.getId();

    }


    private void kafkaSendMsg(MdsInterfaceTask entity, String topicName) {
//        InterfaceTaskKafkaVO kafkaVO = EntityUtils.transform(entity,InterfaceTaskKafkaVO.class);
//        KaInterfacetaskMsg message = new KaInterfacetaskMsg();
//
//        message.setId(entity.getId());
//        message.setInterfaceTaskKafkaVO(kafkaVO);

        log.info("【{} 生产信息】  message = {}",topicName, JSONUtil.toJsonStr(entity));
        //3，放入消息队列中
        try {
            hibaseQueueUtil.offer(true,topicName, entity.getId());
        }catch (Exception e){
            log.error("【Redis-Queue】 插入redis QUEUE 失败 {} | 数据 : {}",topicName,entity.getId());
            hibaseTaskUtil.taskFail(entity.getId(), ExceptionUtil.getRootCauseMessage(e));
        }
//        kafkaTemplate.send(topicName, new Gson().toJson(message));
//        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topicName, new Gson().toJson(message));
//        try {
//            SendResult<String, String> stringStringSendResult = future.get();
//
//        } catch (InterruptedException | ExecutionException e) {
//            e.printStackTrace();
//        }
//        return entity.getId();
    }

    /**
     * 从缓存/数据库查询type表信息
     * @param interfaceTaskVO
     * @return
     */
    private MdsInterfaceType getInterfaceType(MdsInterfaceTaskThirdCallVO interfaceTaskVO){


        //1，缓存中获取type表信息，无记录则从数据库取
        if ( StrUtil.isBlank(interfaceTaskVO.getBusinessTypeCode())){
            log.info("该业务类型code为空: {}",interfaceTaskVO);
            throw new HibaseException("该业务类型code为空");
        }
        MdsInterfaceType type;
        type  = mdsInterfaceTypeService.getOne(new QueryWrapper<MdsInterfaceType>().eq("business_type_code", interfaceTaskVO.getBusinessTypeCode()));
        //判断启用状态，不是启用状态则返回
        if (type ==null) {
            //如果查询的数据不存在，插入key为null
            log.info("业务类型code不存在: {}",type);
            throw new HibaseException(999,"业务类型code不存在");
        }
        if ( null ==type.getBeActive() || !type.getBeActive()){
            log.info("该业务类型code已经被禁用: {}",type);
            throw new HibaseException(999,"该业务类型code已经被禁用");
        }

        return type;
    }

    /**
     * 拼装task表记录并入库
     * @param interfaceTaskVO
     * @param entity
     * @param interfaceType
     */
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    @Override
    public void saveTask(MdsInterfaceTaskThirdCallVO interfaceTaskVO, MdsInterfaceTask entity, MdsInterfaceType interfaceType) {
        entity.setBusinessId(interfaceTaskVO.getBusinessId());
        entity.setBusinessCode(interfaceTaskVO.getBusinessName()==null?interfaceTaskVO.getBusinessCode():interfaceTaskVO.getBusinessName());
        entity.setBusinessTypeCode(interfaceType.getBusinessTypeCode());
        entity.setRequestData(interfaceTaskVO.getRequestData());
        entity.setReceiverSysUrl(interfaceType.getUrl());
        entity.setBusinessType(interfaceType.getBusinessTypeName());
        entity.setSenderSys(interfaceType.getSenderSys());
        entity.setReceiverSys(interfaceType.getReceiverSys());
        entity.setDealStatus(TaskDealStatus.TASK_DEAL_READY.getName());
        entity.setExecTime(DateUtil.date());
        super.store(entity);
    }

    @Override
    public void ignore(List<String> ids, String remarks) {
        List<MdsInterfaceTask> taskList = (List<MdsInterfaceTask>) super.listByIds(ids);
        boolean isAllFail = taskList.stream().allMatch(p -> StrUtil.equalsAny(p.getDealStatus(),TaskDealStatus.TASK_DEAL_FAIL.getName(),TaskDealStatus.TASK_DEAL_READY.getName()));
        HibaseAssert.isTrue(isAllFail,"请选择失败或准备状态的接口任务");

        taskList.forEach(p->{
            p.setDealStatus(TaskDealStatus.TASK_DEAL_IGNORE.getName());
            p.setRemarks(remarks);
        });
        super.storeBatch(taskList);
        log.info("【TASK-IGNORE】 忽略接口任务 ids :{} | remarks :{} ",ids,remarks);

    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void failedRetry(MdsInterfaceTask mdsInterfaceTask) {
        MdsInterfaceType type  = mdsInterfaceTypeService.getOne(new QueryWrapper<MdsInterfaceType>().eq("business_type_code", mdsInterfaceTask.getBusinessTypeCode()));
        if(ObjectUtil.isEmpty(type)){
            log.error("[TASK-RETRY] 重试任务 类型 {} 不存在,请配置 | 数据 : {}",mdsInterfaceTask.getBusinessTypeCode(),mdsInterfaceTask.getId());
            return;
        }
        mdsInterfaceTask.setRetryCount(mdsInterfaceTask.getRetryCount()+1);
        mdsInterfaceTask.setDealStatus(TaskDealStatus.TASK_DEAL_READY.getName());
        this.store(mdsInterfaceTask);
        TransactionSynchronizationManager.registerSynchronization( //在一个有事务的方法中，等事务提交后调另外一个方法可以用
                new TransactionSynchronizationAdapter() {
                    @Override
                    public void afterCommit() {
                        try {
                            hibaseQueueUtil.offer(type.getTopicNameNormal(), mdsInterfaceTask.getId());
                        }catch (Exception e){
                            log.error("【Redis-Queue】 插入redis QUEUE 失败 {} | 数据 : {}",type.getTopicNameNormal(),mdsInterfaceTask.getId());
                            hibaseTaskUtil.taskFail(mdsInterfaceTask.getId(), ExceptionUtil.getRootCauseMessage(e));
                        }
                    }
                }
        );

    }

    /**
     * 直接请求url不走消息队列
     * @param interfaceTaskVO
     */
    @Override
    public String syncCall(MdsInterfaceTaskThirdCallVO interfaceTaskVO) {
        MdsInterfaceTask entity = new MdsInterfaceTask();

        //1，缓存中获取type表信息
        MdsInterfaceType interfaceType = getInterfaceType(interfaceTaskVO);
        //2，拼装task表记录并入库
        this.saveTask(interfaceTaskVO, entity, interfaceType);
        //3，直接调用url
        MdsInterfaceTask save =  execUrl(entity);
        entity.setResponseData(save.getResponseData());
        entity.setDealStatus(save.getDealStatus());
        entity.setRetryCount(save.getRetryCount());
        //4，更新数据库状态
        super.saveOrUpdate(entity);
        return entity.getResponseData();
    }

    /**
     * 调用第三方接口
     *
     * @param task
     */
    @Override
    public MdsInterfaceTask execUrl(MdsInterfaceTask task) {
        //用于更新响应报文
        MdsInterfaceTask entity = new MdsInterfaceTask();

        entity.setRetryCount(task.getRetryCount());
        entity.setExecTime(DateUtil.date());

        //设置对象头，否则会报415 null错误
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);

        if (StrUtil.isNotEmpty(task.getRequestData())) {

            HttpEntity<Object> httpEntity = new HttpEntity<>(task.getRequestData(),headers);
            HibaseAssert.isNull(task.getReceiverSysUrl(),"service.mds.interfacetask.impl.mdsInterfaceTaskServiceImpl.urlNotNull");
            ResponseEntity<String> responseEntity = restTemplate.exchange(task.getReceiverSysUrl(), HttpMethod.POST, httpEntity, String.class);
            HibaseAssert.isNull(responseEntity,"service.mds.interfacetask.impl.mdsInterfaceTaskServiceImpl.requestFail");

            //String post = HttpUtil.post("http://10.48.3.192:8004/mds-interface-task/syncCall", JSONUtil.toJsonStr(interfaceTaskVo));

            ResponseInfo info = JSON.parseObject(responseEntity.getBody(), ResponseInfo.class);
            HibaseAssert.isNull(info,"service.mds.interfacetask.impl.mdsInterfaceTaskServiceImpl.deserializationFail");

            if (responseEntity.getBody() != null && responseEntity.getStatusCodeValue() == 200 && "0".equals(info.getCode())) {
                //调用，成功更新数据库状态
                entity.setId(task.getId());
                entity.setResponseData(responseEntity.getBody());
                entity.setDealStatus(TaskDealStatus.TASK_DEAL_SUCCESS.getName());

            }else {
                entity.setId(task.getId());
                entity.setDealStatus(TaskDealStatus.TASK_DEAL_FAIL.getName());
                //如果为null则表明是第一次调用，将其设置为0
                if (entity.getRetryCount() == null) {
                    entity.setRetryCount(0);
                }else {
                    //不为null则是重试，需要将重试次数+1
                    entity.setRetryCount(entity.getRetryCount() + 1);
                }
                entity.setResponseData(responseEntity.getBody());

            }
        }

        return entity;
    }

    @Override
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public MdsInterfaceTask storeByNewTransactional(MdsInterfaceTask task) {
        return super.store(task);
    }

    /**
     * 查询接口任务表列表
     */
    @Override
    public PageInfo<MdsInterfaceTask> selectMdsInterfaceTaskList(MdsInterfaceTaskSearchVO searchVO) {
        QueryWrapper<MdsInterfaceTask> wrapper = new QueryWrapper();
        wrapper.eq(StrUtil.isNotBlank(searchVO.getBusinessId()), "business_id" ,searchVO.getBusinessId());
        wrapper.eq(StrUtil.isNotBlank(searchVO.getSenderSys()), "sender_sys" ,searchVO.getSenderSys());
        wrapper.eq(StrUtil.isNotBlank(searchVO.getReceiverSys()), "receiver_sys" ,searchVO.getReceiverSys());
        wrapper.eq(StrUtil.isNotBlank(searchVO.getBusinessType()), "business_type" ,searchVO.getBusinessType());
        wrapper.eq(StrUtil.isNotBlank(searchVO.getBusinessCode()), "business_code" ,searchVO.getBusinessCode());
        wrapper.eq(StrUtil.isNotBlank(searchVO.getDealStatus()), "deal_status" ,searchVO.getDealStatus());
        wrapper.eq(StrUtil.isNotBlank(searchVO.getBusinessTypeCode()), "business_type_code" ,searchVO.getBusinessTypeCode());
        if (StrUtil.isNotBlank(searchVO.getKeyWord())){
            wrapper.or().eq("business_type" ,searchVO.getKeyWord());
            wrapper.or().eq("business_code" ,searchVO.getKeyWord());
            wrapper.or().eq("business_type_code" ,searchVO.getKeyWord());
        }
        wrapper.orderByDesc("id");
        wrapper.select("id", "business_type_code","sender_sys","receiver_sys","business_type","business_id","business_code","deal_status","retry_count","exec_time"
                ,"create_by","create_time","update_by","update_time","remarks"); 
        return super.page(searchVO, wrapper);
    }



}